# -*- coding: utf-8 -*-
from time import sleep
from datetime import timedelta, datetime
from utils.operators.spark_submit import SparkSubmitOperator
from jms.ods.dep.yl_jms_tms_effective_state import jms_dep__yl_jms_tms_effective_state
from jms.time_sensor import time_after_01_00

jdbcUrl = '{{ var.json.oracle_tms_pro.url }}'
username = '{{ var.json.oracle_tms_pro.username }}'
password = '{{ var.json.oracle_tms_pro.password }}'
nowdt = '{{ execution_date | cst_ds }}'
nextdt = '{{ execution_date | date_add(1) | cst_ds }}'
table = "yl_tms_branch_time_effective"
env = '{{ var.value.env_sync }}'

jsonpara = """{
"reader":{
"connect":{
"url":"jdbcUrlpara",
"username":"usernamepara",
"password":"passwordpara",
"driver":"oracle.jdbc.driver.OracleDriver"
},
"dbtype":"oracle",
"tableName":"tablepara",
"where":"1=1",
"query":"",
"splitColumn":"",
"equalitySectioning":1,
"containsnull":0,
"fetchsize":"1000",
"threadNumber":1
},
"channel":{
"filterAbnormalCharacter":1
},
"writer":{
"dbtype":"hive",
"tableName":"tablepara",
"database":"bidefault",
"writeMode": "overwrite",
"partitionColumn":"dt",
"partitionValue":"nowdt"},
"settting":{
"env":"envpara"}
}""".replace("jdbcUrlpara", jdbcUrl).replace("usernamepara", username).replace("passwordpara", password). \
    replace("nowdt", nowdt).replace("nextdt", nextdt). \
    replace("tablepara", table). \
    replace("envpara", env)


jms_ods__yl_tms_branch_time_effective_test = SparkSubmitOperator(
    task_id='jms_ods__yl_tms_branch_time_effective_test',
    email=['yl_etl@yl-scm.com','yl_bigdata@yl-scm.com'],
    name='jms_ods__yl_tms_branch_time_effective_test_{{ execution_date | date_add(1) | cst_ds }}',
    pool='oracle_tms',
    pool_slots=1,
    execution_timeout=timedelta(hours=1),
    driver_memory='2G',
    executor_memory='2G',
    executor_cores=2,
    num_executors=1,
    java_class='com.yunlu.bigdata.jobs.synchrotool.DataSynchDriver',  # spark 主类
    application='hdfs:///scheduler/jms/spark/sync/spark_sync.jar',  # spark jar 包
    application_args=[jsonpara],
)

# 延迟 20 分钟运行任务
# 等待数据同步至 Oracle
# def pre_execute(*args, **kwargs):
#     tz = jms_ods__yl_tms_branch_time_effective_test.dag.timezone
#     now = tz.convert(datetime.now())
#     should_after = now.replace(hour=0, minute=30, second=0, microsecond=0)
#
#     while now < should_after:
#         # Oracle 尚未完成同步
#         diff = should_after - now
#         print(f'Will be run after {should_after.time()}, now sleep {diff}')
#         sleep(min(60., diff.total_seconds()))
#         now = tz.convert(datetime.now())
#
#     print(f'Ready to run')


# jms_ods__yl_tms_branch_time_effective_test.pre_execute = pre_execute

jms_ods__yl_tms_branch_time_effective_test  << jms_dep__yl_jms_tms_effective_state
